
package cn.sunline.dreamvalue.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * hbase环境控制类<br/>
 * 即负责创建并保存连接等<br/>
 * 单例模式
 * Created by Administrator on 2017/6/21.
 */
@Service
public class HbaseEnv {

    private static final Logger logger = LoggerFactory.getLogger(HbaseEnv.class);

    private static HbaseEnv hbaseEnv = null;

    @Value("${hbase.zookeeper.servers}")
    private  String zookeeperStr;
    @Value("${hbase.zookeeper.port}")
    private String zookeeperPort;
    @Value("${hbase.krb5.conf.path}")
    private String krb5_path;
    @Value("${hbase.znode.parent}")
    private String znode_parent;
    @Value("${hbase.kerberos.keytab.path}")
    private String keytab_path;
    @Value("${hbase.kerberos.username}")
    private String userName;

    private HbaseEnv() {
    }

    //    private Configuration hbaseCfg=null;
    private Connection conn;

    private long lastCreateConnectionTime;

/**
     * 获取一个hbase环境上下文的实例
     *
     * @return 实例
     */

    public static HbaseEnv getInstance() throws IOException {
        if (null == hbaseEnv) {
            synchronized (HbaseEnv.class) {
                hbaseEnv = new HbaseEnv();
                hbaseEnv.init();
            }
        }
        return hbaseEnv;
    }

    private void init() throws IOException {

         System.setProperty("java.security.krb5.conf", krb5_path);
        //System.setProperty("java.security.krb5.realm", "TDH");
        System.setProperty("hadoop.home.dir","/usr/local/hadoop");
        Configuration hbaseCfg = HBaseConfiguration.create();

        hbaseCfg.set("hbase.zookeeper.quorum", zookeeperStr);
        hbaseCfg.set("hbase.zookeeper.property.clientPort", zookeeperPort);
        hbaseCfg.set("zookeeper.znode.parent", znode_parent);

        hbaseCfg.set("hbase.master.kerberos.principal", userName + "/_HOST@TDH");
        hbaseCfg.set("hbase.regionserver.kerberos.principal", userName + "/_HOST@TDH");
        hbaseCfg.set("hbase.security.authentication", "kerberos");
        hbaseCfg.set("hadoop.security.authentication", "kerberos");


        logger.info("........【加载到hbase数据库配置文件】.....keytab_path:"+keytab_path);
        logger.info("........【加载到hbase数据库配置文件】.....krb5_path:"+krb5_path);
        UserGroupInformation.setConfiguration(hbaseCfg);
        UserGroupInformation.loginUserFromKeytab(userName, keytab_path);

        conn = ConnectionFactory.createConnection(hbaseCfg);

        lastCreateConnectionTime = System.currentTimeMillis();

    }



/**
     * 获取一个hbase的连接<br/>
     * 参数配置都在refcfg.properties中
     *
     * @return 返回一个连接
     */

    public Connection getConnection() throws IOException {

        long currTime = System.currentTimeMillis();

        //1000*60*60*9=32400000
        if (null == conn || (currTime - lastCreateConnectionTime > 32400000)) {
            synchronized (HbaseEnv.class) {
                init();
            }
        }

        return conn;
    }
}

